package com.primi.chen.zookeeper;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * Hello world!
 */
@Data
@Slf4j
public class NodeDiscovery implements Runnable, Watcher, DataMonitorListener {

    private ServiceConfiguration configuration;

    private Map<String, NodeInfo> nodeInfos;

    private Boolean connect = true;

    private ZooKeeper zk;

    private Long sessionId;

    private String root;

    private String znode;

    private Thread shutdownHook;

    private CloseListener closeListener;

    private int existCode = 0;

    public static NodeDiscovery of(ServiceConfiguration configuration) {
        NodeDiscovery server = new NodeDiscovery();
        server.setConfiguration(configuration);
        return server;
    }

    private void init() throws Exception {
        // 初始化 zookeeper 连接, 并监听
        this.nodeInfos = new HashMap<>(10);
        this.zk = new ZooKeeper(configuration.getZkservers(), configuration.getSessionTimeout(), this);
        this.sessionId = this.zk.getSessionId();
        this.znode = configuration.filename();
        this.root = "/".concat(configuration.getName());
        ;
        log.info("[ZK] {}", this.sessionId);
        this.checkAndCreateRootDir();
        this.registerRootChinas();
        this.loadChildrenNodes(this.root);
        this.mkdirServerDir(CreateMode.EPHEMERAL_SEQUENTIAL);
    }

    private void registerRootChinas() throws Exception {
        this.zk.getChildren(this.root, this);
    }

    private void mkdirServerDir(CreateMode createMode) throws Exception {
        znode = this.zk.create(znode, configuration.data(), ZooDefs.Ids.OPEN_ACL_UNSAFE, createMode);
        this.zk.exists(znode, this);
    }

    /**
     * 创建 Root 目录
     */
    private void checkAndCreateRootDir() throws KeeperException, InterruptedException {
        if (this.zk.exists(root, false) == null) {
            this.zk.create(root, String.valueOf(this.zk.getSessionId()).getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    @SneakyThrows
    @Override
    public void run() {
        this.init();
        synchronized (this) {
            while (this.connect) {
                log.info("等等待..... {}", this.znode);
                wait();
                log.info("等到 notify 了额");
            }
        }
        this.exit();
    }

    public void close() {
        synchronized (this) {
            this.connect = false;
            this.notifyAll();
        }
    }

    public void exit() {
        try {
            this.zk.close();
            this.closing(existCode);
            if (this.closeListener != null) {
                this.closeListener.close();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @SneakyThrows
    @Override
    public void process(WatchedEvent watchedEvent) {
        log.info("Procesor: {}", watchedEvent.getPath());
        switch (watchedEvent.getType()) {
            case NodeCreated: {
                log.info("注册节点: {}", watchedEvent.getPath());
                return;
            }
            case NodeDeleted: {
                // 节点删除 , 调用 Close ;
                this.whenZNodeDelete(watchedEvent);
                return;
            }
            case NodeDataChanged: {
                // 数据更改
                return;
            }
            case NodeChildrenChanged: {
                // 子节点变化
                this.loadChildrenNodes(watchedEvent);
                return;
            }
            case None:
                log.info("No Any thing happen ");
        }
    }

    private void loadChildrenNodes(WatchedEvent watchedEvent) throws Exception {
        this.loadChildrenNodes(watchedEvent.getPath());
    }

    /**
     * 获取所有节点
     *
     * @param path
     */
    private void loadChildrenNodes(String path) throws Exception {
        if (!this.connect) {
            return;
        }
        List<String> nodes = this.zk.getChildren(path, this).stream().map(node -> path + "/" + node).collect(Collectors.toList());
        for (String node : nodes) {
            Stat stat = this.zk.exists(node, false);
            byte[] data = this.zk.getData(node, false, stat);
            nodeInfos.put(node, NodeInfo.of(node, data));
        }
        log.info("[{}] JSON: \n {}", this.znode, JSON.toJSONString(nodeInfos, SerializerFeature.PrettyFormat, SerializerFeature.WriteMapNullValue, SerializerFeature.WriteDateUseDateFormat));
    }

    private void whenZNodeDelete(WatchedEvent watchedEvent) throws Exception {
        if (StringUtils.equals(watchedEvent.getPath(), this.znode)) {
            // 当自己被删除时
            if (this.connect) {
                log.warn("异常节点关闭... 重新创建");
                this.mkdirServerDir(CreateMode.EPHEMERAL);
            } else {
                log.info("Node 节点删除, ZK 删除 . 关闭应用程序;");
                this.existCode = 0;
                this.close();
            }
        } else {
            // 当其他节点被删除时
            this.nodeInfos.remove(watchedEvent.getPath());
        }
    }

    @Override
    public void exists(byte[] data) {

    }

    @Override
    public void closing(int rs) {
        log.info("关闭... 应用 {}", this.znode);
    }
}
